package com.example.kafka.fast;


import com.example.kafka.constant.KafkaConstant;
import com.example.kafka.constant.StreamConstant;
import org.apache.kafka.common.protocol.MessageUtil;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;

/**
 * @author Liyuhang
 */
public class KafkaStreamFastStart {

    public static void main(String[] args) {
        //1.stream构造器
        StreamsBuilder builder = new StreamsBuilder();

        //2.流式计算
        KafkaStreamDemo1.streamProcess(builder);

        //3.创建kafkaStream对象
        final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(),KafkaStreamDemo1.getDefaultProperties());
        final CountDownLatch latch = new CountDownLatch(1);
        //4.开启stream流式计算
        StreamProducerDemo1 streamProducerDemo = new StreamProducerDemo1(StreamConstant.IN_PUT_TOPIC);
        List<CompletableFuture> futureList = startJob(kafkaStreams,streamProducerDemo);
        futureList.forEach(CompletableFuture::join);
        // attach shutdown handler to catch control-c
        Runtime.getRuntime().addShutdownHook(new Thread("streams-temperature-shutdown-hook") {
            @Override
            public void run() {
                kafkaStreams.close();
                latch.countDown();
            }
        });


    }

    private static List<CompletableFuture> startJob(KafkaStreams kafkaStreams,StreamProducerDemo1 streamProducerDemo) {
        CompletableFuture future1 = CompletableFuture.runAsync(() -> {
            kafkaStreams.start();
        });
        CompletableFuture future2 = CompletableFuture.runAsync(() -> {
            streamProducerDemo.process(KafkaConstant.generatorUuid());
        });
        return Arrays.asList(future1,future2);
    }


}
